package com.ly.ql.RocketMqListener.Produce;

import com.ly.ql.Result.Response.Response;
import com.ly.ql.RocketMqListener.fallback.SendCallbackListener;
import lombok.RequiredArgsConstructor;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;

import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.Map;

@Component
@RequiredArgsConstructor
public class ProduceUtils {
    private final RocketMQTemplate rocketMQTemplate;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.sync-tag}")
    private String syncTag;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.async-tag}")
    private String asyncag;

    @Value(value = "${rocketmq.producer.topic}:${rocketmq.producer.oneway-tag}")
    private String onewayTag;

    /**
     * @Description : 同步像RocketMq发送消
     * @return
     */
    public Response sendMessage(String str, Map<String,Object> map,Long OID){
        MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(str);

        if (map!=null&&map.size()>0){
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();
                stringMessageBuilder.setHeader(key,value);
            }
        }
        stringMessageBuilder.setHeader(RocketMQHeaders.KEYS,OID);
        Message<String> message = stringMessageBuilder.build();
        // 设置发送地和消息信息并发送同步消息
        SendResult sendResult = rocketMQTemplate.syncSend(syncTag, message);
        Response msg = new Response();
        // 解析发送结果
        if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
            msg.setMsg(sendResult.getMsgId() + " : " + sendResult.getSendStatus());
        }
        return msg;
    }

    /**
     * @Description : 异步像RocketMq发送消息
     * @param str
     * @param id
     * @return
     */
    public Response sendSynMessage(String str, Map<String,Object> map,Long OID){
        MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(str);

        if (map!=null&&map.size()>0){
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();
                stringMessageBuilder.setHeader(key,value);
            }
        }
        stringMessageBuilder.setHeader(RocketMQHeaders.KEYS,OID);
        Message<String> message = stringMessageBuilder.build();
        // 设置发送地和消息信息并发送同步消息 异步发送不会管是否发送成功 直接走下一步
        Response msg = new Response();
        try {
            rocketMQTemplate.asyncSend(asyncag, message,new SendCallbackListener(OID));
        }catch (RuntimeException e){
            msg.setCode(500);
            msg.setMsg(e.getMessage().toString());
            return msg;
        }
        // 解析发送结果
       msg.setCode(200);
       msg.setMsg("异步MQ发送成功");
       return msg;
    }
    public Response sendMessageOrder(String str, Map<String,Object> map,Long OID){
        MessageBuilder<String> stringMessageBuilder = MessageBuilder.withPayload(str);

        if (map!=null&&map.size()>0){
            for (Map.Entry<String, Object> entry : map.entrySet()) {
                String key = entry.getKey();
                Object value = entry.getValue();
                stringMessageBuilder.setHeader(key,value);
            }
        }
        stringMessageBuilder.setHeader(RocketMQHeaders.KEYS,OID);
        Message<String> message = stringMessageBuilder.build();
        //rocketMQTemplate.getProducer().setDefaultTopicQueueNums(4); 默认四个队列
        rocketMQTemplate.setMessageQueueSelector(new MessageQueueSelector() {
            @Override
            public MessageQueue select(List<MessageQueue> mqs, org.apache.rocketmq.common.message.Message msg, Object arg) {
                return mqs.get(0);
            }
        });
        // 设置发送地和消息信息并
        SendResult sendResult = rocketMQTemplate.syncSendOrderly(syncTag, message,"sdsdasd");
        Response msg = new Response();
        // 解析发送结果
        if (sendResult.getSendStatus() == SendStatus.SEND_OK) {
            msg.setCode(200);
        }
        return msg;
    }

}
